package com.duoduo.commonn.redis.factory;

import com.duoduo.common.api.constants.Constant;
import com.duoduo.common.api.constants.LogConstant;
import com.duoduo.common.api.domain.DelayExecuteMessage;
import com.duoduo.common.api.exception.delay.DelaySendException;
import com.duoduo.common.api.help.*;
import com.duoduo.common.api.myinterface.delay.DelayManage;
import com.duoduo.common.api.myinterface.mq.MessageHandler;
import com.duoduo.common.api.myinterface.tracer.RedisTracer;
import com.duoduo.common.api.spi.extension.ExtensionLoader;
import com.duoduo.commonn.redis.LogThreadLocal;
import com.duoduo.commonn.redis.handle.TraceHandler;
import com.duoduo.commonn.redis.myinterface.RedisOperator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.Serializable;
import java.util.Set;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class RedisDelayFactory implements DelayManage {

    private static Logger logger = LoggerFactory.getLogger(RedisDelayFactory.class);
    private RedisOperator redisOperator;
    private static final String DELAY_JOB_ENV = PropertiesInnerUtil.getString(Constant.DELAY_PROP_LOCATION, Constant.DELAY_PROP_ENV);
    private static final String DELAY_JOB_PRE = DELAY_JOB_ENV == null ? "delayJ:" : "delayJ:" + DELAY_JOB_ENV;
    private static final String DELAY_BUCKET = DELAY_JOB_ENV == null ? "delayBucket" : "delayBucket:" + DELAY_JOB_ENV;
    private static final String DELAY_UPDATE = DELAY_JOB_ENV == null ? "lock" : "lock:" + DELAY_JOB_ENV;
    private static final int DELAY_BUCKET_ONCE_GET_MAX_COUNT = 100;
    private static final String DELAY_UPDATE_THREADNAME = "delayup";
    private static final String TASKFLAG = PropertiesInnerUtil.getString(Constant.DELAY_PROP_LOCATION, Constant.DELAY_PROP_TASKFLAG);

    public RedisDelayFactory() {
        //延迟任务的redis实现，redis默认使用jedis操作
        String redisImplType = SpringContextInnerUtil.get(Constant.REDIS_IMPL);
        logger.info(LogConstant.REDIS_CACHE + " impl type: {}", redisImplType);
        // 如果有链路跟踪
     	if(SpringContextInnerUtil.containsBean(Constant.TRACER_BEANNAME)){
	        RedisOperator proxy = ExtensionLoader.getExtensionLoader(RedisOperator.class).getExtension(redisImplType);
	        RedisTracer tracer = ExtensionLoader.getExtensionLoader(RedisTracer.class).getExtension(Constant.TRACER_REDIS_CLIENT_NAME);
	        redisOperator = DynamicProxyInnerUtil.getProxy(proxy, new TraceHandler<RedisOperator>(tracer));
     	} else {
     		redisOperator = ExtensionLoader.getExtensionLoader(RedisOperator.class).getExtension(redisImplType);
     	}
    }
    
	@Override
    public <T extends Serializable> void sendDelayMessage(DelayExecuteMessage<T> message, int delayTime) throws DelaySendException {
    	try{
			// 把任务加入redis
	    	addJobToPool(message, delayTime);
    	} catch (Exception e){
    		throw new DelaySendException(LogConstant.REDIS_DELAY + " addJobToPool error: ", e);
    	}
    	try{
    		// 把任务加入zset
	    	addToBucket(message, delayTime);
    	} catch (Exception e){
    		throw new DelaySendException(LogConstant.REDIS_DELAY + " addToBucket error: ", e);
    	}
    }
    
    /**任务加入redis，key是traceId标识唯一任务；value是DelayExecuteMessage，有效期是延迟执行时间+60s
     * @param message
     * @param delayTime
     */
    private <T extends Serializable> void addJobToPool(DelayExecuteMessage<T> message, int delayTime) {
    	delayTime = delayTime + Constant.DELAY_REDIS_TIME;
    	redisOperator.setImplSeri(Constant.REDIS_PROP_DEFAULT_PRE_NAME, DELAY_JOB_PRE, String.valueOf(message.getTraceId()), message, delayTime);
    }
    
    /**任务加入zset，value是traceId，排序标识是过期时间倒序，就是最快过期的排在最上面
     * @param message
     * @param delayTime
     */
    private <T extends Serializable> void addToBucket(DelayExecuteMessage<T> message, int delayTime) {
    	double score = System.currentTimeMillis() + (delayTime * 1000l);
    	redisOperator.addWithSortedSet(Constant.REDIS_PROP_DEFAULT_PRE_NAME, message.getExecuteKey(), DELAY_BUCKET, score, String.valueOf(message.getTraceId()));
    }
    
    @Override
	public void init(String dealKey, int minThreadNum, int maxThreadNum, int queueCapacity, String threadName, MessageHandler handler) throws Exception {
    	ExecutorService es = new ThreadPoolExecutor(minThreadNum, maxThreadNum, 60L, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(queueCapacity), new NamingThreadFactory(threadName));
    	
    	ConsumeWorker consumeWorker = new ConsumeWorker(dealKey, es, handler);
    	Thread t = new Thread(consumeWorker);
    	t.setName(DELAY_UPDATE_THREADNAME);
    	t.start();
		
    	if(logger.isDebugEnabled()){
    		logger.debug(LogConstant.REDIS_DELAY + " pools and dealhandles init, dealKey: {}", dealKey);
    	}
    }
    
    private class ConsumeWorker implements Runnable {

		private String dealKey;
		private ExecutorService es;
		private MessageHandler handler;

		public ConsumeWorker(String dealKey, ExecutorService es, MessageHandler handler) {
			super();
			this.dealKey = dealKey;
			this.es = es;
			this.handler = handler;
		}

		@Override
		public void run() {
			LogThreadLocal.restrainLockLog.set(true);
			while (true){
		    	try{
		    		Set<String> traceIds = null;
					try {
						// 这里用redis的加锁方式，主要是因为这是基于redis实现的延迟任务
						boolean lockSuccess = LockOperatorInnerUtil.tryGetDistributedLock(Constant.LOCK_DEFAULT_PRE_NAME, TASKFLAG, DELAY_UPDATE, null, 0, false);
						if(lockSuccess) {
							traceIds = redisOperator.rangeByScoreWithSortedSet(Constant.REDIS_PROP_DEFAULT_PRE_NAME, dealKey, DELAY_BUCKET, DateInnerUtil.getTodayLong(), 0, 0, DELAY_BUCKET_ONCE_GET_MAX_COUNT);
							if(ListInnerUtil.isNotEmpty(traceIds)){
								redisOperator.zrem(Constant.REDIS_PROP_DEFAULT_PRE_NAME, dealKey, DELAY_BUCKET, traceIds);
							}
						}
					}catch (Exception e) {
						logger.error(LogConstant.REDIS_DELAY + " redisDelay update add lock error", e);
					}finally {
						try {
							LockOperatorInnerUtil.releaseDistributedLock(Constant.LOCK_DEFAULT_PRE_NAME, TASKFLAG, DELAY_UPDATE, true);
						}catch (Exception e) {
							logger.error(LogConstant.REDIS_DELAY + " redisDelay update release lock error", e);
						}
					}
			    	if(ListInnerUtil.isNotEmpty(traceIds)){
			    		for (String traceId : traceIds) {
			    			Object result = redisOperator.pop(Constant.REDIS_PROP_DEFAULT_PRE_NAME, DELAY_JOB_PRE, traceId);
			    			if(result != null){
			    				es.execute(() -> {
			    					try {
			                			handler.execute((byte[])result);
			                		} catch (Exception e) {
			                			String msg = LogConstant.REDIS_DELAY + " consumer error, dealKey: " + dealKey;
			                			logger.error(msg, e);
			                		}
			    				});
			    			}
						}
			    	}else{
			    		//短暂休眠，避免可能的活锁
		                try {
							Thread.sleep(1000);
						} catch (InterruptedException e) {
							// no-op
						}
			    	}
		    	} catch (Exception e) {
		    		// 这里有可能是redis出问题了
	    			logger.error(LogConstant.REDIS_DELAY + " consumer unknow error, dealKey: " + dealKey, e);
		    		//短暂休眠
	                try {
						Thread.sleep(1000);
					} catch (InterruptedException e1) {
						e1.printStackTrace();
					}
		    	}
	    	}
		}
		
    }
    
}